package gpool

import (
	"sync"
	"sync/atomic"
)

type Worker interface {
	Work()
}

const (
	un_closed = iota
	in_closed
)

// GoroutinePool
// 线程池,功能:
// 1.控制线程池的状态(1.开始;2.终止)
// 2.动态控制线程中子线程的运行数量
type GoroutinePool struct {
	locker     sync.RWMutex
	wg         sync.WaitGroup
	isClosed   int32
	closeChan  chan bool // 线程池开关
	addChan    chan bool
	reduceChan chan bool
	maxRunning int64 // 最大运行数
	running    int64 // 正在运行数
}

func InitGoroutinePool(maxRunning int64) *GoroutinePool {
	return &GoroutinePool{
		maxRunning: maxRunning,
		running:    0,
		isClosed:   un_closed,
		closeChan:  make(chan bool),
		addChan:    make(chan bool),
		reduceChan: make(chan bool),
		wg:         sync.WaitGroup{},
		locker:     sync.RWMutex{},
	}
}

func (p *GoroutinePool) IsClosed() bool {
	p.locker.RLock()
	defer p.locker.RUnlock()
	return p.isClosed == in_closed
}

func (p *GoroutinePool) Close() {
	atomic.StoreInt32(&p.isClosed, in_closed)
	p.locker.Lock()
	defer p.locker.Unlock()
	select {
	case <-p.closeChan: //Already closed. Don't close again.
	default:
		p.closeChan <- true
		close(p.closeChan)
	}
}

func (p *GoroutinePool) Pause() {
	p.ResetMaxRunning(0)
}

func (p *GoroutinePool) Start(w Worker) {
	p.addRunning(p.maxRunning)
T:
	for {
		select {
		case <-p.closeChan:
			break T
		case <-p.addChan:
			go p.runGo(w)
		}
	}
	//thinkLog.InfoLog.Println("ending")
	p.wg.Wait()
	//thinkLog.InfoLog.Println("end")
	close(p.addChan)
	close(p.reduceChan)
}

func (p *GoroutinePool) runGo(w Worker) {
	select {
	case <-p.reduceChan:
		return
	default:
	}
	defer func() {
		if !p.IsClosed() {
			p.addChan <- true
		}
	}()
	p.wg.Add(1)
	defer p.wg.Done()
	atomic.AddInt64(&(p.running), 1)
	defer atomic.AddInt64(&(p.running), -1)
	//fmt.Printf("go running:%v \t max:%v", p.running, p.maxRunning)
	w.Work()

}

func (p *GoroutinePool) ResetMaxRunning(maxNew int64) {
	n := maxNew - p.maxRunning
	p.maxRunning = maxNew
	switch {
	case n == 0:
		return
	case n < 0:
		p.reduceRunning(-n)
	case n > 0:
		p.addRunning(n)
	}
}

func (p *GoroutinePool) addRunning(n int64) {
	go func() {
		defer func() {
			if r := recover(); r != nil {
			}
		}()
		for i := int64(0); i < n; i++ {
			//thinkLog.InfoLog.Println("add <-", p.IsClosed())
			if p.IsClosed() {
				return
			}
			p.addChan <- true
			//thinkLog.InfoLog.Println("add <- done")
		}
	}()

}

func (p *GoroutinePool) reduceRunning(n int64) {
	go func() {
		defer func() {
			if r := recover(); r != nil {
			}
		}()
		for i := int64(0); i < n; i++ {
			//thinkLog.InfoLog.Println("reduce <-", p.IsClosed())
			if p.IsClosed() {
				return
			}
			p.reduceChan <- true
			//thinkLog.InfoLog.Println("reduce <- done")
		}
	}()

}
